package com.leyantech.ecp.agenx.session.manager.handler;

import static com.leyantech.ecp.agenx.common.config.ConfigManager.ASSISTANT_EVENT_TOPIC;

import com.leyantech.chaos.kafka.KafkaProducerStringClient;
import com.leyantech.ecp.agenx.common.handler.ActionHandler;
import com.leyantech.ecp.agenx.common.message.EnumWsFnType;
import com.leyantech.ecp.agenx.common.redis.RedissonOperation;
import com.leyantech.ecp.agenx.common.redis.entity.OnlineAssistantSession;
import com.leyantech.ecp.agenx.common.service.StoreHomeService;
import com.leyantech.ecp.agenx.session.client.Assistant;
import com.leyantech.storehome.AssistantSeatModeInfo;
import com.leyantech.storehome.ModeExtendInfo;

import com.google.common.base.Preconditions;
import com.google.inject.Inject;
import com.google.inject.Singleton;
import io.vertx.core.json.JsonObject;
import lombok.extern.slf4j.Slf4j;
import org.apache.commons.lang3.StringUtils;

import java.util.ArrayList;
import java.util.Collections;
import java.util.Comparator;

/**
 * @author eahau, {@literal <eahau@foxmail.com>}
 * @date 2022-11-07.
 */
@Slf4j
@Singleton
public class AssistantLoginHandler extends ActionHandler {

  private final StoreHomeService storeHomeService;

  private final KafkaProducerStringClient kafkaClient;

  @Inject
  public AssistantLoginHandler(
      StoreHomeService storeHomeService,
      KafkaProducerStringClient kafkaClient) {
    Preconditions.checkArgument(StringUtils.isNotBlank(ASSISTANT_EVENT_TOPIC.get()));
    this.storeHomeService = storeHomeService;
    this.kafkaClient = kafkaClient;
  }

  @Override
  public EnumWsFnType actionType() {
    return EnumWsFnType.ASSISTANT_LOGIN;
  }

  @Override
  public void handle(JsonObject message) {

    log.info("received assistant LOGIN event {}", message);

    final JsonObject header = message.getJsonObject("header");

    final long timestamp = header.getLong("ts");

    final Assistant assistant = Assistant.fromJson(message.getJsonObject("body").toString());

    assistant.setTimestamp(timestamp);
    assistant.setAction(actionType().type);

    updateOnlineAssistantSession(assistant);

    // 通知其它服务客服登录登出.
    final String key = String.join(".", assistant.getStoreId(), assistant.getAssistantId());

    kafkaClient.send(ASSISTANT_EVENT_TOPIC.get(), key, assistant.toString());

    final String nodeId = assistant.getNodeId();
    RedissonOperation.NODE_ASSISTANT.executor(nodeId).run(set -> {
      if (set.add(key)) {
        log.info("add assistant {} to node {} succeed", key, nodeId);
      }
    }, "add");
  }

  private void updateOnlineAssistantSession(Assistant assistant) {
    final String sid = assistant.getStoreId();
    final String aid = assistant.getAssistantId();

    final ModeExtendInfo modeExtendInfo = storeHomeService
        .getStoreCustomReplyMode(sid, aid)
        .orElse(Collections.emptyList())
        .stream()
        .findFirst()
        .map(AssistantSeatModeInfo::getModeList)
        .flatMap(list -> list.stream().min(Comparator.comparingInt(ModeExtendInfo::getOrder)))
        .orElse(ModeExtendInfo.getDefaultInstance());

    final String serviceMode = modeExtendInfo.getCode();
    final ArrayList<String> functions = new ArrayList<>(modeExtendInfo.getFunctionList());

    assistant.setServiceMode(serviceMode);
    assistant.setFunctionList(functions);

    final OnlineAssistantSession onlineAssistantSession = OnlineAssistantSession.builder()
        .storeId(sid)
        .assistantId(aid)
        .assistantName(assistant.getAssistantName())
        .clientUuid(assistant.getClientUuid())
        .clientVersion(assistant.getClientVersion())
        .functions(functions)
        .serviceMode(serviceMode)
        .timestamp(assistant.getTimestamp())
        .action(assistant.getAction())
        .nodeId(assistant.getNodeId())
        .build();

    RedissonOperation.STORE_ASSISTANT
        .executor(sid)
        .run(map -> map.put(aid, onlineAssistantSession), "assistantLogin");
  }

}
